package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2023/2/2
 *
 * 找到超时(30min)和对账失败(只有支付信息没有订单信息)的订单。
 *
 * 34729,create,,1558430842
 * 34729,pay,sd76f87d6,1558430844
 *
 *
 *      四种情况:
 *          1. create --- pay ,未超时   正常
 *          2. create --- pay ,超时。 pay距离create 超过30min  异常
 *          3. create   只下单，不付款。 正常
 *          4. pay    对账失败，没有找到订单下单的信息  异常
 *
 *          两次行为有时间限制时，就用会话窗口。
 *              gap = 30 min
 *
 *        思路：  create 先到 ----- pay后到
 *              ①按照orderId 分组
 *              ②开一个30min的 session窗口
 *              ③窗口计算时，如果窗口中有2个数据，说明是情况1，正常
 *                      如果只有1个数据
 *                          判断如果是create ,把自己写入状态
 *                          如果是pay ，从状态中尝试读 create，读到，是情况2
 *                                                         读不到，是情况4
 */
public class Demo6_OrderPay
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy
            .<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getEventTime());


        env.readTextFile("data/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>()
            {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] words = value.split(",");

                    return new OrderEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3]) * 1000
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
           .keyBy(OrderEvent::getOrderId)
           .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
           .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>()
           {

               private ValueState<String> orderId;

               @Override
               public void open(Configuration parameters) throws Exception {
                   orderId = getRuntimeContext().getState(new ValueStateDescriptor<>("orderId", String.class));
               }

               @Override
               public void process(Long oid, Context context, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {

                   List<OrderEvent> orderEvents = MyUtil.parseList(elements);

                   if (orderEvents.size() == 1){

                       OrderEvent orderEvent = orderEvents.get(0);

                       //当前是create
                       if ("create".equals(orderEvent.getEventType())){

                           //可能是3或2，具体需要后续看是否有pay到达判断
                           orderId.update(orderEvent.getOrderId().toString());

                       }else{

                           //是pay
                           //从状态中取，如果可以取到数据，说明当前订单的create已经到了，超时
                           String orderStr = this.orderId.value();

                           if (orderStr != null){
                               out.collect(oid + "超时...");
                           }else{
                               //取不到，是缺少create
                               out.collect(oid + "只有pay...");
                           }
                       }

                   }

               }
           })
           .print().setParallelism(1);

        env.execute();




    }
}
